package com.gitee.andy.rpc.client;

import java.lang.reflect.Proxy;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;

import org.springframework.beans.factory.InitializingBean;

import com.gitee.andy.rpc.core.MessageDecoder;
import com.gitee.andy.rpc.core.MessageEncoder;
import com.gitee.andy.rpc.entity.RPC;
import com.gitee.andy.rpc.entity.RpcRequest;
import com.gitee.andy.rpc.entity.RpcResponse;
import com.gitee.andy.util.Clazzs;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class RpcClient implements InitializingBean {

	private Bootstrap bootstrap;
	private EventLoopGroup group;
	private String pkgName;

	private Map<Class<?>, Object> beans = new HashMap<>();

	private String inetHost;

	private int inetPort;

	public void shutdown() {
		group.shutdownGracefully();
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		RpcClientHandler rpcClientHandler = new RpcClientHandler();
		Set<Class<?>> clazzes = Clazzs.getClasses(pkgName);
		clazzes.stream().filter(clazz -> {
			return clazz.isInterface() && clazz.isAnnotationPresent(RPC.class);
		}).forEach(clazz -> {
			beans.put(clazz,
					Proxy.newProxyInstance(clazz.getClassLoader(), new Class<?>[] { clazz }, (proxy, method, args) -> {
						return rpcClientHandler.write(new RpcRequest(fetchNextSeq(), method, args)).getResult();
					}));
		});
		group = new NioEventLoopGroup(1);
		bootstrap = new Bootstrap();
		bootstrap.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true)
				.handler(new ChannelInitializer<SocketChannel>() {
					@Override
					public void initChannel(SocketChannel ch) throws Exception {
						ch.pipeline().addLast(new MessageDecoder(RpcResponse.class), new MessageEncoder(),
								rpcClientHandler);
					}
				});
		bootstrap.connect(inetHost, inetPort).sync().channel();
	}

	private static AtomicLong nextSeq = new AtomicLong();

	private static long fetchNextSeq() {
		return nextSeq.incrementAndGet();
	}

	public <T> T get(Class<T> clazz) {
		return (T) beans.get(clazz);
	}

	public String getPkgName() {
		return pkgName;
	}

	public void setPkgName(String pkgName) {
		this.pkgName = pkgName;
	}

	public String getInetHost() {
		return inetHost;
	}

	public void setInetHost(String inetHost) {
		this.inetHost = inetHost;
	}

	public int getInetPort() {
		return inetPort;
	}

	public void setInetPort(int inetPort) {
		this.inetPort = inetPort;
	}

}
